# -*- coding:utf-8 -*-

from myinit import spark, sc
from logger import worklog
from config import config
import pymysql

host = config.MysqlHost
port = config.MysqlPort
user = config.MysqlUser
password = config.MysqlPassword
database = config.MysqlDatabase
#dbTable = config.MysqlTable

def into_MySQL(sparkDF, dbTable, mode='overwrite'):
    '''
    写入MySQL数据库。
    连接MySQL数据库的相关参数已写死(根据实际需求改连接参数)
    '''
    conn = pymysql.connect(host=host, user=user, password=password, port=int(port))
    cursor = conn.cursor()
    cursor.execute("CREATE DATABASE IF NOT EXISTS {} default character set utf8;".format(database))
    
    # 写入MySQL
    try:
        worklog.info("准备写入 MySQL 数据库")
        sparkDF.write.jdbc(
            url=
            """jdbc:mysql://{0}:{1}/{2}?user={3}&password={4}&useUnicode=true&characterEncoding=utf-8
                &useServerPrepStmts=false&rewriteBatchedStatements=true""".format(host,port,database,user,password),
            mode=mode,
            table=dbTable,
            properties={"driver": 'com.mysql.cj.jdbc.Driver'})
        worklog.info("成功写入 MySQL 数据库")
    except:
        sparkDF.write.jdbc(
            url=
            """jdbc:mysql://{0}:{1}/{2}?user={3}&password={4}&useUnicode=true&characterEncoding=utf-8
                &useServerPrepStmts=false&rewriteBatchedStatements=true""".format(host,port,database,user,password),
            mode=mode,
            table=dbTable,
            properties={"driver": 'com.mysql.jdbc.Driver'})
        worklog.info("成功写入 MySQL 数据库")
